package simple.utils.tracking;

import com.google.common.base.Preconditions;
import com.google.protobuf.ByteString;
import com.google.protobuf.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import simple.utils.data.Bytes;
import simple.utils.data.IterableKeyValue;
import simple.utils.data.KeyValue;
import simple.utils.data.SeekingIterator;
import simple.utils.data.serializers.*;
import simple.utils.tracking.storage.KeyValueStorageFactory;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by martin on 5.5.14.
 */
public class Tracking {

    private static final Logger LOGGER = LoggerFactory.getLogger(Tracking.class);
    private static final Map<String, Tracking> instances = new ConcurrentHashMap<>();
    private static final ReentrantLock classLock = new ReentrantLock();
    private static volatile KeyValueStorageFactory defaultStorageFactory;
    private static final Map<Class, Serializer> serializers = new ConcurrentHashMap<>();
    private static final Set<Class> unknownSerializers = Collections.newSetFromMap(new ConcurrentHashMap<Class, Boolean>());

    static {
        // int
        serializers.put(int.class, new IntegerSerializer());
        serializers.put(Integer.class, new IntegerSerializer());
        // long
        serializers.put(long.class, new LongSerializer());
        serializers.put(Long.class, new LongSerializer());
        // double
        serializers.put(double.class, new DoubleSerializer());
        serializers.put(Double.class, new DoubleSerializer());
        // boolean
        serializers.put(boolean.class, new BooleanSerializer());
        serializers.put(Boolean.class, new BooleanSerializer());
        // byte
        serializers.put(Byte.class, new ByteSerializer());
        serializers.put(byte.class, new ByteSerializer());
        // short
        serializers.put(short.class, new ShortSerializer());
        serializers.put(Short.class, new ShortSerializer());
        // float
        serializers.put(float.class, new FloatSerializer());
        serializers.put(Float.class, new FloatSerializer());
        // char
        serializers.put(char.class, new CharSerializer());
        serializers.put(Character.class, new CharSerializer());
        // string
        serializers.put(String.class, new StringSerializer());
        // enum
        serializers.put(Enum.class, new EnumSerializer());
        // generic gpb serializer
        serializers.put(Message.class, new GPBMessageSerializer());

    }

    private String trackingName;
    private final ReentrantLock instanceLock = new ReentrantLock();
    private final KeyValueStorageFactory storageFactory;


    private Tracking(String name, KeyValueStorageFactory factory) {
        this.trackingName = name;
        this.storageFactory = factory;
    }


    /**
     * @param name
     * @return returns instance of tracking object. Creates new instance with default levelDB storage if this instance was not previously registered
     */
    public static Tracking getInstance(String name) {
        Preconditions.checkNotNull(name);
        Tracking tracking = instances.get(name);
        if (tracking == null) {
            classLock.lock();
            try {
                tracking = instances.get(name);
                if (tracking == null) {
                    Preconditions.checkNotNull(defaultStorageFactory, "Default storage factory is NULL, must set it first!");
                    tracking = new Tracking(name, defaultStorageFactory);
                    instances.put(name, tracking);
                }
            } finally {
                classLock.unlock();
            }
        }
        return tracking;
    }


    public static void registerSerializer(Class c, Serializer s) {
        serializers.put(c, s);
    }

    public static void setDefaultStorageFactory(KeyValueStorageFactory storageFactory) {
        Tracking.defaultStorageFactory = storageFactory;
    }

    public static void createStorage(String name, KeyValueStorageFactory factory) {
        try {
            classLock.lock();
            if (instances.get(name) == null) {
                instances.put(name, new Tracking(name, factory));
            }
        } finally {
            classLock.unlock();
        }
    }

    /**
     * @param trackedObjectType type of tracked object
     * @param trackedObjectId   id of trackeg object
     * @param action            action executed on tracked object
     * @param dataNames
     * @param data
     */
    public void track(String trackedObjectType, String action, String trackedObjectId, String[] dataNames, Object[] data, long ts) {
        try {
            // get storage
            KeyValue<byte[], TrackingProto.TrackingRecord> db = storageFactory.get(trackedObjectType);
            // check data names and data fields
            if (dataNames == null && data == null) {
                dataNames = new String[0];
                data = new Object[0];
            }

            Preconditions.checkArgument(dataNames.length == data.length, "Data names length must match data length");
            // create builder
            TrackingProto.TrackingRecord.Builder bl = TrackingProto.TrackingRecord.newBuilder()
                    .setAction(action)
                    .setTs(ts);
            // serialize data (key, type, value)
            for (int i = 0; i < dataNames.length; i++) {
                Preconditions.checkNotNull(data[i], "Data object is null.");
                String name = dataNames[i];
                Class<?> cls = data[i].getClass();
                Object o = data[i];
                Serializer s = serializers.get(cls);

                if (s == null) {
                    for (Class c : serializers.keySet()) {
                        if (c.isAssignableFrom(cls)) {
                            s = serializers.get(c);
                            serializers.put(cls, s);
                            unknownSerializers.remove(cls);
                            break;
                        }
                    }
                }

                if (s == null) {
                    if (!unknownSerializers.contains(cls)) {
                        LOGGER.warn("Unknown serializer for class " + o.getClass() + ", serializing to JSON");
                        unknownSerializers.add(cls);
                    }
                    s = new JsonObjectSerializer(cls);
                }

                byte[] bytes = s.toBytes(o);
                bl.addData(TrackingProto.Data.newBuilder()
                                .setName(name)
                                .setType(cls.getName())
                                .setData(ByteString.copyFrom(bytes))
                                .build()
                );
            }
            // put to DB
            db.put(TrackingUtils.generateID(trackedObjectId, ts), bl.build());
        } catch (Throwable t) {
            LOGGER.error("Error tracking event objectType: " + trackedObjectType + ", objectId: " + trackedObjectId + ", action: " + action + ", dataNames" + Arrays.toString(dataNames) + ", data: " + Arrays.toString(data), t);
        }
    }

    public static <E> E deserialize(Class<E> cls, TrackingProto.Data d) {
        Object o = deserialize(d);
        return (E) o;
    }

    public static Object deserialize(TrackingProto.Data d) {
        String type = d.getType();
        Object o = null;
        try {
            Class<?> cl = Class.forName(type);
            Serializer serializer = serializers.get(cl);
            o = serializer.fromBytes(d.getData().toByteArray());
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }

        return o;
    }

    public void track(String trackedObjectType, String action, String trackedObjectId, String[] dataNames, Object[] data) {
        track(trackedObjectType, action, trackedObjectId, dataNames, data, System.currentTimeMillis());
    }

    public void track(Enum trackingObject, String action, String trackedObjectId, String[] keys, Object[] values) {
        track(trackingObject.name(), action, trackedObjectId, keys, values);

    }

    public void track(String trackingObject, Enum action, String trackedObjectId, String[] keys, Object[] values) {
        track(trackingObject, action.name(), trackedObjectId, keys, values);

    }

    public void track(Enum trackingObject, Enum action, String trackedObjectId, String[] keys, Object[] values) {
        track(trackingObject.name(), action.name(), trackedObjectId, keys, values);
    }

    public void track(Enum trackingObject, String action, String trackedObjectId, String[] keys, Object[] values, long ts) {
        track(trackingObject.name(), action, trackedObjectId, keys, values, ts);

    }

    public void track(String trackingObject, Enum action, String trackedObjectId, String[] keys, Object[] values, long ts) {
        track(trackingObject, action.name(), trackedObjectId, keys, values, ts);

    }

    public void track(Enum trackingObject, Enum action, String trackedObjectId, String[] keys, Object[] values, long ts) {
        track(trackingObject.name(), action.name(), trackedObjectId, keys, values, ts);
    }

    public TrackingProto.TrackingRecords getActions(String trackedObjectType, String id) {
        IterableKeyValue<byte[], TrackingProto.TrackingRecord> db = storageFactory.get(trackedObjectType);
        SeekingIterator<byte[], TrackingProto.TrackingRecord> it = db.iterator();

        TrackingProto.TrackingRecords.Builder bl = TrackingProto.TrackingRecords.newBuilder();

        try {
            byte[] startBytes = TrackingUtils.generateStartBytes(id);
            it.seek(startBytes);

            while (it.hasNext()) {
                Map.Entry<byte[], TrackingProto.TrackingRecord> entry = it.next();
                if (TrackingUtils.keyMatches(entry.getKey(), id)) {
                    bl.addRecords(entry.getValue());
                } else {
                    break;
                }
            }
        } finally {
            try {
                it.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return bl.build();
    }

    public TrackingProto.TrackingRecords getActions(String trackedObjectType, String id, long start, long end) {
        IterableKeyValue<byte[], TrackingProto.TrackingRecord> db = storageFactory.get(trackedObjectType);
        SeekingIterator<byte[], TrackingProto.TrackingRecord> it = db.iterator();

        TrackingProto.TrackingRecords.Builder bl = TrackingProto.TrackingRecords.newBuilder();

        try {
            byte[] startBytes = TrackingUtils.generateStartBytes(id);
            ByteArrayOutputStream bos = new ByteArrayOutputStream();
            bos.write(startBytes);
            bos.write(Bytes.longToBytes(start));
            it.seek(bos.toByteArray());

            while (it.hasNext()) {
                Map.Entry<byte[], TrackingProto.TrackingRecord> entry = it.next();
                if (TrackingUtils.keyMatchesRange(entry.getKey(), id, end)) {
                    bl.addRecords(entry.getValue());
                } else {
                    break;
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                it.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return bl.build();
    }

    public TrackingProto.TrackingRecords getActions(Enum trackedObjectType, String id) {
        return getActions(trackedObjectType.name(), id);
    }
}
